package useTask.hbaseTask;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class scanHbase {
    static Configuration conf;
    static Connection HConn;
    static Table table;

    private static Scan scan = new Scan();
    private static FilterList filterList = new FilterList();


    // 全表查询
    public static void scan(String tableName) throws IOException {
        conf = HBaseConfiguration.create();
        HConn = ConnectionFactory.createConnection(conf);
        table = HConn.getTable(TableName.valueOf(tableName));

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String row_key = Bytes.toString(CellUtil.cloneRow(cell));
                String Family = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(row_key + " " + Family + " " + column + " " + value);
            }
            System.out.println("------------------------------------");
        }
        // 释放资源
        scanner.close();
        table.close();
        HConn.close();
    }

    // 行建查询
    public static void scan(String tableName, String rowKey) throws IOException {
        conf = HBaseConfiguration.create();
        HConn = ConnectionFactory.createConnection(conf);
        table = HConn.getTable(TableName.valueOf(tableName));

        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new BinaryComparator(rowKey.getBytes()));
        scan.setFilter(rowFilter);

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String row_key = Bytes.toString(CellUtil.cloneRow(cell));
                String Family = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(row_key + " " + Family + " " + column + " " + value);
            }
            System.out.println("------------------------------------");
        }
        // 释放资源
        scanner.close();
        table.close();
        HConn.close();
    }

    // 查询指定列柱、指定列中等于指定值的数据
    public static void scan(String tableName, String family,String column,String value) throws IOException {
        conf = HBaseConfiguration.create();
        HConn = ConnectionFactory.createConnection(conf);
        table = HConn.getTable(TableName.valueOf(tableName));

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                family.getBytes(), column.getBytes(),
                CompareOperator.EQUAL, value.getBytes()
        );
        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String row_key = Bytes.toString(CellUtil.cloneRow(cell));
                String Family = Bytes.toString(CellUtil.cloneFamily(cell));
                String Column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String Value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(row_key + " " + Family + " " + Column + " " + Value);
            }
            System.out.println("------------------------------------");
        }
        // 释放资源
        scanner.close();
        table.close();
        HConn.close();
    }

    // 查询指定列柱、指定列中大于等于指定值的数据
    public static void scanOver(String tableName, String family,String column,String value) throws IOException {
        conf = HBaseConfiguration.create();
        HConn = ConnectionFactory.createConnection(conf);
        table = HConn.getTable(TableName.valueOf(tableName));

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                family.getBytes(), column.getBytes(),
                CompareOperator.GREATER_OR_EQUAL, value.getBytes()
        );
        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String row_key = Bytes.toString(CellUtil.cloneRow(cell));
                String Family = Bytes.toString(CellUtil.cloneFamily(cell));
                String Column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String Value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(row_key + " " + Family + " " + Column + " " + Value);
            }
            System.out.println("------------------------------------");
        }
        // 释放资源
        scanner.close();
        table.close();
        HConn.close();
    }

    // 查询指定列柱、指定列中小于等于指定值的数据
    public static void scanLess(String tableName, String family,String column,String value) throws IOException {
        conf = HBaseConfiguration.create();
        HConn = ConnectionFactory.createConnection(conf);
        table = HConn.getTable(TableName.valueOf(tableName));

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                family.getBytes(), column.getBytes(),
                CompareOperator.LESS_OR_EQUAL, value.getBytes()
        );
        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String row_key = Bytes.toString(CellUtil.cloneRow(cell));
                String Family = Bytes.toString(CellUtil.cloneFamily(cell));
                String Column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String Value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(row_key + " " + Family + " " + Column + " " + Value);
            }
            System.out.println("------------------------------------");
        }
        // 释放资源
        scanner.close();
        table.close();
        HConn.close();
    }

}
